package com.atguigu.day03;

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
import org.apache.flink.util.Collector;

import java.util.Random;

public class Example7 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env.addSource(new RichParallelSourceFunction<Integer>() {
            private Boolean running = true;
            private final Random random = new Random();
            @Override
            public void run(SourceContext<Integer> ctx) throws Exception {
                while (running){
                    ctx.collect(random.nextInt(10000));
                }
            }

            @Override
            public void cancel() {
                running = false;
            }
        }).keyBy(x->true)
                .process(new KeyedProcessFunction<Boolean, Integer, Double>() {
                    private ValueState<Tuple2<Integer,Integer>> acc;
                    private ValueState<Long> timeTs;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        acc = getRuntimeContext().getState(new ValueStateDescriptor<Tuple2<Integer, Integer>>(
                                "acc", Types.TUPLE(Types.INT,Types.INT)
                        ));
                        timeTs= getRuntimeContext().getState(new ValueStateDescriptor<Long>(
                                "timeTS",Types.LONG
                        ));
                    }

                    @Override
                    public void processElement(Integer value, Context ctx, Collector<Double> out) throws Exception {
                        if (acc.value()==null){
                            acc.update(Tuple2.of(value,1));
                            Tuple2<Integer, Integer> value1 = acc.value();
                        }else {
                            Tuple2<Integer, Integer> value1 = acc.value();
                            acc.update(Tuple2.of(acc.value().f0+value,acc.value().f1+1));
                        }
                        if (timeTs.value()==null){
                            timeTs.update(ctx.timerService().currentProcessingTime()+5*1000);
                            ctx.timerService().registerProcessingTimeTimer(timeTs.value());
                        }
                    }

                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<Double> out) throws Exception {
                        super.onTimer(timestamp, ctx, out);
                        out.collect((double)acc.value().f0/acc.value().f1);
                        timeTs.update(null);
                    }
                }).print();

        env.execute();
    }
}
